package com.rexense.beacon.service;

import com.rexense.beacon.entity.Msg;
import lombok.Getter;
import lombok.Setter;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@ConfigurationProperties(prefix = "beacon.msg")
@Service
public class MsgService {

    @Autowired
    private RedissonClient redisson;

    @Autowired
    private LocationService locationService;

    @Setter
    @Getter
    private long waitingMillis = 2000;

    @Setter
    @Getter
    private long sleepMillis = 300;

    private final static String MSG_KEY_SET = "MSG:KEY:SET:";
    private final static String MSG_KEY_PRIORITY_QUEUE = "MSG:KEY:PRIORITY_QUEUE:";
    private final static String MSG_KEY_SORTEDSET_PREFIX = "MSG:KEY:SORTEDSET:";

    private final static ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(20, 30, 1, TimeUnit.HOURS, new LinkedBlockingQueue<>());

    public void receive(Msg msg) {
        //保证msgId唯一,所有的msgId放入set中
        RSet<String> msgKeySet = redisson.getSet(MSG_KEY_SET);
        if (!msgKeySet.contains(msg.key())) {
            RLock lock = msgKeySet.getLock(msg.key());
            lock.lock();
            if (!msgKeySet.contains(msg.key())) {
                msgKeySet.add(msg.key());

                //放入任务队列
                RPriorityQueue<Msg.MsgKey> priorityQueue = redisson.getPriorityQueue(MSG_KEY_PRIORITY_QUEUE);
                priorityQueue.trySetComparator(new Msg.MsgIdComparator());
                priorityQueue.add(msg.toMsgKey());
            }
            lock.unlock();
        }

        //消息放入SORTED SET
        RSortedSet<Msg> set = redisson.getSortedSet(MSG_KEY_SORTEDSET_PREFIX + msg.key());
        set.add(msg);
    }

    @PostConstruct
    public void run() {
        EXECUTOR.execute(() -> {
            try {
                while (true) {
                    long now = System.currentTimeMillis();
                    RPriorityQueue<Msg.MsgKey> priorityQueue = redisson.getPriorityQueue(MSG_KEY_PRIORITY_QUEUE);
                    RLock lock = redisson.getLock(MSG_KEY_PRIORITY_QUEUE + "LOCK");
                    lock.lock();
                    Msg.MsgKey msgKey = priorityQueue.peek();
                    if (msgKey != null && now - msgKey.getTimestamp() > waitingMillis) {
                        System.out.println(msgKey);
                        priorityQueue.poll();
                        lock.unlock();
                        EXECUTOR.execute(new LocationTask(msgKey.msgKey));
                    } else {
                        lock.unlock();
                        TimeUnit.MILLISECONDS.sleep(sleepMillis);
                    }
                }
            } catch (Exception ignored) {
            }
        });
    }

    class LocationTask implements Runnable {

        private final String msgKey;

        public LocationTask(String msgKey) {
            this.msgKey = msgKey;
        }

        @Override
        public void run() {
            RSortedSet<Msg> sortedSet = redisson.getSortedSet(MSG_KEY_SORTEDSET_PREFIX + msgKey);
            Collection<Msg> msgs = sortedSet.readAll();
            if (msgs != null && !msgs.isEmpty()) {
                sortedSet.removeAll(msgs);
                List<Msg> msgList = new ArrayList<>(msgs);
                locationService.locate(msgList);
            }
        }
    }

}
